package com.pig4cloud.pig.common.core.config;

import com.bblocks.enums.QueueBaseEnum;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.retry.ImmediateRequeueMessageRecoverer;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

/**
 * Rabbit启动配置
 * 1、spring.rabbitmq.host需要配置才会启动
 * 2、配置中需要排除mq默认启动配置
 * “
 * spring:
 *   autoconfigure:
 *     exclude: org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration
 * ”
 *
 * 参考：https://blog.csdn.net/weixin_62108279/article/details/131451083 ，可以再优化下
 *
 * @description:消息队列配置
 * @author: ZHaojin
 * @time: 2023/8/16 下午3:53
 */
//@ConditionalOnBean(name = "spring.rabbitmq.host")
@ConditionalOnProperty(name = "spring.rabbitmq.host", matchIfMissing = false)
@Configuration
@Slf4j
@Getter
public class TopicRabbitMqConfig extends RabbitAutoConfiguration {
	@Value("${spring.application.name}")
	private String applicationName;


	/**
	 * 当前mq消费线程
	 */
	@Value("${spring.rabbitmq.listener.simple.concurrency:15}")
	private int DEFAULT_CONCURRENT ;

	/**
	 * 最大值
	 */
	@Value("${spring.rabbitmq.listener.simple.max-concurrency:15}")
	private int DEFAULT_CONCURRENT_MAX ;

	/**
	 *  #消费者ack模式
	 *         #NONE模式，则只要收到消息后就立即确认（消息出列，标记已消费），有丢失数据的风险
	 *         #AUTO模式，看情况确认，如果此时消费者抛出异常则消息会返回到队列中
	 *         #MANUAL模式，需要显式的调用当前channel的basicAck方法
	 */
	@Value("${spring.rabbitmq.listener.simple.acknowledge-mode:AUTO}")
	private String acknowledgeMode ;

	/**
	 * 监听重试是否可用(默认false ，会导致消息消费不完整-本处调整默认值为默认)
	 */
	@Value("${spring.rabbitmq.listener.simple.retry.enabled:false}")
	private boolean retryEnabled ;

	/**
	 * #最大重试次数 默认为3
	 */
	@Value("${spring.rabbitmq.listener.simple.retry.max-attempts:5}")
	private Integer retryMaxAttempts;

	/**
	 * # 传递消息的时间间隔 默认1000ms
	 */
	@Value("${spring.rabbitmq.listener.simple.retry.initial-interval:1000}")
	private Integer retryInitialInterval ;

	/**
	 * 应用前一次间隔乘法器，默认1
	 * 以上配置消息间隔：0,1,2,4,8，16 (s)
	 */
	@Value("${spring.rabbitmq.listener.simple.retry.multiplier:2}")
	private Integer retryMultiplier ;


	/**
	 * 重试失败，处理：默认（RejectAndDontRequeueRecoverer），本处重试失败
	 * 1、RejectAndDontRequeueRecoverer 默认拒绝并且不会将消息重新发回队列
	 * 2、ImmediateRequeueMessageRecoverer 重入队列（本处会进入继续循环消费），错误的信息会一直执行
	 * 3、RepublishMessageRecoverer 将消息重新发送到指定队列
	 *
	 * 可以n次结果失败：
	 * 1、写入延时队列
	 * 2、写入db，直接结束
	 *
	 * @return
	 */
	@Bean
	public MessageRecoverer messageRecoverer(){
		return new ImmediateRequeueMessageRecoverer();
	}

	/**
	 * 配置mq的执行并发
	 * 配置如下：
	 * @RabbitListener(queues = {"xxx"},
	 *                     containerFactory = "customContainerFactory")
	 * @param configurer
	 * @param connectionFactory
	 * @return
	 */
	@Bean("customContainerFactory")
	public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
																 ConnectionFactory connectionFactory
			,MessageRecoverer messageRecoverer) {

		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		factory.setConcurrentConsumers(DEFAULT_CONCURRENT);
		factory.setMaxConcurrentConsumers(DEFAULT_CONCURRENT_MAX);
		configurer.configure(factory, connectionFactory);
		if("AUTO".equals(acknowledgeMode)){
			factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
		}

		factory.setMessageConverter(jackson2JsonMessageConverter());
		// 设置通知调用链 （这里设置的是重试机制的调用链）
		if(retryEnabled){
			factory.setAdviceChain(
					org.springframework.amqp.rabbit.config.RetryInterceptorBuilder
							.stateless()
							.recoverer(messageRecoverer)
							.retryOperations(rabbitRetryTemplate())
							.build()
			);
		}

		return factory;
	}

    /**
     * 定义消息转换实例
     * @return
     */
    @Bean
    MessageConverter jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

	@Bean
	public RetryTemplate rabbitRetryTemplate() {
		RetryTemplate retryTemplate = new RetryTemplate();

		// 设置监听  调用重试处理过程
		retryTemplate.registerListener(new RetryListener() {
			@Override
			public <T, E extends Throwable> boolean open(RetryContext retryContext, RetryCallback<T, E> retryCallback) {
				// 执行之前调用 （返回false时会终止执行）

				return true;
			}

			@Override
			public <T, E extends Throwable> void close(RetryContext retryContext, RetryCallback<T, E> retryCallback, Throwable throwable) {
				// 重试结束的时候调用 （最后一次重试 ）
				if(throwable != null)
					log.error("---------------最后一次调用:",throwable);

				return ;
			}
			@Override
			public <T, E extends Throwable> void onError(RetryContext retryContext, RetryCallback<T, E> retryCallback, Throwable throwable) {
				//  异常 都会调用
				log.error("-----第{}次调用,异常：{}",retryContext.getRetryCount(),throwable.getMessage());
			}
		});
		retryTemplate.setBackOffPolicy(backOffPolicyByProperties());
		retryTemplate.setRetryPolicy(retryPolicyByProperties());
		return retryTemplate;
	}

	@Bean
	public ExponentialBackOffPolicy backOffPolicyByProperties() {
		ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
		//long maxInterval = properties.getListener().getSimple().getRetry().getMaxInterval().getSeconds();
		long initialInterval = this.getRetryInitialInterval();
		double multiplier = this.getRetryMultiplier();
		// 重试间隔
		backOffPolicy.setInitialInterval(initialInterval);
		// 重试最大间隔
		//backOffPolicy.setMaxInterval(maxInterval * 1000);
		// 重试间隔乘法策略
		backOffPolicy.setMultiplier(multiplier);
		return backOffPolicy;
	}

	@Bean
	public SimpleRetryPolicy retryPolicyByProperties() {
		SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
		int maxAttempts = this.getRetryMaxAttempts();
		retryPolicy.setMaxAttempts(maxAttempts);
		return retryPolicy;
	}



	 /**
	 * 以下为支付相关
	 */
	@Bean
	TopicExchange payResultTopic(){
		return (TopicExchange) ExchangeBuilder.topicExchange(QueueBaseEnum.QUEUE_PAY_RESULT.getExchange())
				.durable(true)
				.build();
	}
	@Bean(name = "payResultQueue")
	public Queue payResultQueue() {
		return new Queue(QueueBaseEnum.QUEUE_PAY_RESULT.getName()+"."+applicationName);
	}

	@Bean
	public Binding directBindingPayResult(TopicExchange payResultTopic, Queue payResultQueue){
		return BindingBuilder.bind(payResultQueue).to(payResultTopic)
				.with(QueueBaseEnum.QUEUE_PAY_RESULT.getRouteKey());//topic.yuezhu.pay.result.#
	}

	/**
	 * 以下为订单取消了 通知相关
	 */
	@Bean
	TopicExchange commOrderCancelTopic(){
		return (TopicExchange) ExchangeBuilder.topicExchange(QueueBaseEnum.QUEUE_COMMORDER_CANCEL.getExchange())
				.durable(true)
				.build();
	}
	@Bean(name = "commOrderCancelQueue")
	public Queue commOrderCancelQueue() {
		return new Queue(QueueBaseEnum.QUEUE_COMMORDER_CANCEL.getName()+"."+applicationName);
	}

	@Bean
	public Binding directBindingCommOrderCancel(TopicExchange commOrderCancelTopic, Queue commOrderCancelQueue){
		return BindingBuilder.bind(commOrderCancelQueue).to(commOrderCancelTopic)
				.with(QueueBaseEnum.QUEUE_COMMORDER_CANCEL.getRouteKey());//topic.yuezhu.pay.result.#
	}





	/**
	 * yuezhu通用交换机
	 */
	@Bean("directExchange")
	public DirectExchange commDirectExchange() {
		return new DirectExchange(QueueBaseEnum.NEEDS_STATUSCHANGED.getExchange(),true,false);
	}

}
